/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cloudy;

import cloudy.tools.DateFmt;
import kafka.examples.KafkaProperties;
import kafka.examples.Producer;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.storm.utils.Utils;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.ExecutionException;

public class CloudyProducer extends Thread {
    private final KafkaProducer<Integer, String> producer;
    private final String topic;
    private final Boolean isAsync;
    volatile int i=0;

    public CloudyProducer(String topic, Boolean isAsync) {
        Properties props = new Properties();
        props.put("bootstrap.servers",
                KafkaProperties.KAFKA_SERVER_URL + ":"
                        + KafkaProperties.KAFKA_SERVER_PORT);
        props.put("client.id", "test");
        props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<>(props);
        this.topic = topic;
        this.isAsync = isAsync;
    }

    public void run() {

        while (true) {
            i++;
            Random random = new Random();
            String[] order_amt = {"10.10", "20.10", "20.2", "50.2", "70.1", "80.1"};
            //String[] area_id = {"1", "2", "3", "4", "5","6", "7", "8"};
            String[] area_id = {"1", "2", "3", "4", "5","6", "7", "8"};
            String messageStr=i+"\t"+order_amt[random.nextInt(5)]+"\t"+ DateFmt.getCountDate(null,DateFmt.date_long)
                    +"\t"+area_id[random.nextInt(8)];
            if (isAsync) { // Send asynchronously
                producer.send(new ProducerRecord<>(topic,messageStr), new DemoCallBack( messageStr));
            } else { // Send synchronously
                try {
                    producer.send(new ProducerRecord<>(topic,messageStr)).get();
                    System.out.println("Sent producer: ( " + messageStr + ")");
                } catch (InterruptedException | ExecutionException e) {
                    e.printStackTrace();
                }
            }
            Utils.sleep(200);
            if(i==100){
                break;
            }
        }
    }
}

class DemoCallBack implements Callback {

    private final String message;

    public DemoCallBack( String message) {
        this.message = message;
    }

    /**
     * A callback method the user can implement to provide asynchronous handling of request completion. This method will
     * be called when the record sent to the server has been acknowledged. Exactly one of the arguments will be
     * non-null.
     *
     * @param metadata  The metadata for the record that was sent (i.e. the partition and offset). Null if an error
     *                  occurred.
     * @param exception The exception thrown during processing of this record. Null if no error occurred.
     */
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if (metadata != null) {
            System.out.println(
                    "message(" +  message + ") sent to partition(" + metadata.partition() +
                            "), " +
                            "offset(" + metadata.offset() + ") in "  );
        } else {
            exception.printStackTrace();
        }
    }

    public static void main(String[] args) {
        boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync");
        CloudyProducer producerThread = new CloudyProducer(KafkaProperties.ORDER_TOPIC, isAsync);
        producerThread.start();
    }

    /**
     * num[["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:1","2017-09-30","1",43],
     * ["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:2","2017-09-30","2",47],
     * ["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:3","2017-09-30","3",44],["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:4","2017-09-30","4",44],["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:5","2017-09-30","5",55],["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:6","2017-09-30","6",null],["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:7","2017-09-30","7",null],["2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8","2017-09-30:8","2017-09-30","8",null]]
     2017-09-30:1 2017-09-30:2 2017-09-30:3 2017-09-30:4 2017-09-30:5 2017-09-30:6 2017-09-30:7 2017-09-30:8
     */
}
